package com.qs.rocketmq.sc.consumer;

import com.qs.rocketmq.sc.service.MySink;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Component;

/**
 * 消费端消费消息处理逻辑
 *
 * @since 2021-9-11 21:00:44
 * @author qinypeng
 */
@Component
public class ScConsumer {

    private static final Logger logger = LoggerFactory.getLogger(ScConsumer.class);

    /*@StreamListener(value = MySink.myInput, condition = "headers['rocketmq_TAGS'] == 'youku'")
    public void onMessage(@Payload Message message) {
        logger.info("消费端线程编号:{} message：{}", Thread.currentThread().getId(), message);
    }*/

    @StreamListener(MySink.myInput)
    public void onMessage(@Payload Message message) {
        logger.info("消费端线程编号:{} message：{}", Thread.currentThread().getId(), message);
    }


   /* @StreamListener(MySink.myInput)
    public void onMessage(@Payload MessageDTO message) {
        logger.info("消费端线程编号:{} 消息内容：{}", Thread.currentThread().getId(), message);
        //throw new RuntimeException("这是一个故意抛出来的异常");
    }*/

    /**
     * 指定topic和消费组，处理异常
     */
    @ServiceActivator(inputChannel = "SC_TEST_TOPIC.sc_consumer_group.errors")
    public void handleError(ErrorMessage errorMessage){
        logger.error("[handleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        logger.error("[handleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        logger.error("[handleError][headers：{}]", errorMessage.getHeaders());
    }

    /**
     * 全局异常处理（errorChannel）
     */
    @StreamListener(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME)
    public void globalHandleError(ErrorMessage errorMessage){
        logger.error("[globalHandleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        logger.error("[globalHandleError][originalMessage：{}]", errorMessage.getOriginalMessage());
        logger.error("[globalHandleError][headers：{}]", errorMessage.getHeaders());
    }

}
